//
// Created by ta&to on 2022/8/22.
//

#include "tp_thread.h"

volatile long taskid = 0;

static inline void *thread_cycle(void *data);

//或群读锁
static inline int thread_rwlock_rdlock(pthread_rwlock_t *lock)
{
    int err;
    err = pthread_rwlock_rdlock(lock);
    if(err == 0)
        return thread_ok;
    return thread_err;
}
//或群写锁
static inline int thread_rwlock_wrlock(pthread_rwlock_t *lock)
{
    int err;
    err = pthread_rwlock_wrlock(lock);
    if(err == 0)
        return thread_ok;
    return thread_err;
}
// 释放读写锁
static inline int thread_rwlock_unlock(pthread_rwlock_t *lock)
{
    int err;
    err = pthread_rwlock_unlock(lock);
    if(err == 0)
        return thread_ok;
    return thread_err;
}
// 获取线程锁
static inline int thread_mutex_lock(pthread_mutex_t *lock)
{
    int err;
    err = pthread_mutex_lock(lock);
    if(err == 0)
        return thread_ok;
    return thread_err;
}
// 释放线程锁
static inline int thread_mutex_unlock(pthread_mutex_t *lock)
{
    int err;
    err = pthread_mutex_unlock(lock);
    if(err == 0)
        return thread_ok;
    return thread_err;
}


static inline int thread_get_threads_num(thread_pool_t* tpt)
{
    thread_rwlock_rdlock(&tpt->thread_rwlock);
    int num = tpt->threads;
    (void)  thread_rwlock_unlock(&tpt->thread_rwlock);
    return num;

}

static inline void thread_add_threads_num(thread_pool_t* tpt, int num){
    thread_rwlock_wrlock(&tpt->thread_rwlock);
    atomicIncr(tpt->threads, num);
    (void) thread_rwlock_unlock(&tpt->thread_rwlock);
}

static inline noder thread_rb_sel_node(Leader *leader, void *key)
{
    noder node = RBtreenodesel(leader, key);
    return node;
}
static inline int thread_rb_add_node(thread_pool_t *tpt, Leader *leader, void *key, void *value)
{
    addlinklisttable(&tpt->list_thread, key);
    RBtreenodeadd(leader, key, value);
    atomicIncr(tpt->threads, 1);
}
static inline int thread_rb_del_node(thread_pool_t *tpt, Leader *leader, void *key, table_t *t)
{
    destrylinklist(&tpt->list_thread, t);
    RBtreenodedel(leader, key);
    atomicDecr(tpt->threads, 1);
}
static inline int thread_rb_del_node_safe(thread_pool_t *tpt, Leader *leader, void *key, table_t *t)
{
    thread_rwlock_wrlock(&tpt->thread_rwlock);
    thread_rb_del_node(tpt, leader, key, t);
    (void) thread_rwlock_unlock(&tpt->thread_rwlock);
}
static inline int thread_rb_mod_node(Leader *leader, void *key, void *value)
{
    treenodemodify(leader, key, value);
}

/**
 * 创建一个任务队列
 * @param ttq 任务队列
 */
void Thread_Create_task_queue(thread_task_queue_t *ttq)
{
    ttq->count = 0;
    ttq->first = NULL;
    ttq->last = &(ttq->first);
}

/*
 * 创建一个任务容器
 */
static inline thread_task_t *newtaskcontainer(void (*handler)(void *data))
{
    thread_task_t *ttt;
    ttt = (thread_task_t *)tp_calloc(sizeof(thread_task_t));
    ttt->next = NULL;
    ttt->handler = handler;
    ttt->taskid = taskid;
    atomicIncr(taskid, 1);
}
/**
 * 向队列中加入任务
 * @param ttq
 * @param task
 */
static inline void Thread_Insert_Task(thread_task_queue_t *ttq, thread_task_t *task)
{
    if(!ttq || !task) return;
    if(ISNULL(ttq->first))
        ttq->first = task;
    *(ttq->last) = task;
    ttq->last = &task->next;
    atomicIncr(ttq->count, 1);
}

static inline int Thread_Insert_Task_Container(thread_pool_t* tpt, void (*handler)(void *data), void *args)
{
    thread_task_t *tt;
    if(NULL != (tt = newtaskcontainer(handler)))
    {
        tt->ctx = args;
        Thread_Insert_Task(&tpt->queue, tt);
        return thread_ok;
    }
    return thread_err;
}


/**
 * 向线程池队列插入任务
 * @param tpt
 * @param handler
 * @param monopoly 是否需要独占 1是 0 否
 * @return
 */
static inline int Thread_Insert_Task_To_Pool(thread_pool_t* tpt, void (*handler)(void *data), void *args)
{
    thread_rwlock_wrlock(&tpt->task_rwlock);
    if(tpt->max_queue>0 && tpt->waiting > tpt->max_queue){
        (void) thread_rwlock_unlock(&tpt->task_rwlock);
        printf(stderr, "task queue is full!");
        return thread_err;
    }
    if(!Thread_Insert_Task_Container(tpt, handler, args))
    {
        (void) thread_rwlock_unlock(&tpt->task_rwlock);
        return thread_err;
    }
    atomicIncr(tpt->waiting, 1);
    pthread_cond_signal(&tpt->cond);
    (void) thread_rwlock_unlock(&tpt->task_rwlock);
    return thread_ok;
}
// 任务入队
int Thread_Task_Post(thread_pool_t* tpt, void (*handler)(void *data), void *args)
{
    if(ISNULL(tpt)) return thread_err;
    return Thread_Insert_Task_To_Pool(tpt, handler, args);
}

static inline thread_task_t* Thread_Get_Task_From_Pool(thread_pool_t* tpt)
{
    thread_task_t *task = tpt->queue.first;
    if(ISNOTNULL(task))
    {
        tpt->queue.first = task->next;
        if (ISNULL(tpt->queue.first)) {
            (&tpt->queue)->last = &(&tpt->queue)->first;
        }
        atomicDecr(tpt->queue.count, 1);
    }
    return task;
}
// 任务出队
thread_task_t* Thread_Task_Pop(thread_pool_t* tpt)
{
    thread_task_t *task;
    if(ISNULL(tpt)) return NULL;
    if(!thread_rwlock_wrlock(&tpt->task_rwlock))        // 虽然是读取数据但是由于牵扯到指针移动所以使用写锁
        return NULL;
    task = Thread_Get_Task_From_Pool(tpt);
    if(ISNOTNULL(task))
        atomicDecr(tpt->waiting, 1);
    thread_rwlock_unlock(&tpt->task_rwlock);
    return task;
}

static inline void destrytaskcontainer(thread_pool_t* tpt, thread_task_t *ttt)
{
    if(ttt) free(ttt);
    ttt = NULL;
}

/*
 * 销毁任务
 */
int Thread_Task_Destry(thread_pool_t* tpt, thread_task_t *ttt)
{
    if(ISNULL(tpt) || ISNULL(ttt)) return NULL;
    if(!thread_rwlock_wrlock(&tpt->task_rwlock))        // 虽然是读取数据但是由于牵扯到指针移动所以使用写锁
        return thread_err;
    destrytaskcontainer(tpt, ttt);
    thread_rwlock_unlock(&tpt->task_rwlock);
    return thread_ok;
}

//任务队列中是否有数据
static inline int hasTask(thread_pool_t *tpt)
{
    if(!thread_rwlock_rdlock(&tpt->task_rwlock))
        return thread_err;
    if(ISNOTNULL(tpt->queue.first))
    {
        thread_rwlock_unlock(&tpt->task_rwlock);
        return thread_ok;
    }
    thread_rwlock_unlock(&tpt->task_rwlock);
    return thread_err;
}
/*
 * 获取任务数
 */
static inline void *logic_computer_task_queue_number(thread_task_t *pp, int *pnum)
{
    if(ISNOTNULL(pp))
        (*pnum)++;
    return pp;
}

static inline long Thread_Get_Task_Number_By_Logic(thread_task_queue_t *ttq)
{
    int num =0;
    thread_task_t *pp;
    if((pp = ttq->first) == NULL) return 0;
    num++;
    while (logic_computer_task_queue_number(pp->next, &num) != NULL && (pp = pp->next) == pp);
    return num;
}
// 通过队列内容获取任务数（不建议使用）
static inline long Thread_Get_Task_Number_From_Pool_By_Logic(thread_pool_t *tpt)
{
    long num = 0;
    thread_rwlock_rdlock(&tpt->task_rwlock);
    num = Thread_Get_Task_Number_By_Logic(&tpt->queue);
    (void) thread_rwlock_unlock(&tpt->task_rwlock);
    return num;
}
static inline long Thread_Get_Task_Number(thread_task_queue_t *ttq)
{
    return ttq->count;
}
// 获取任务数
static inline long Thread_Get_Task_Number_From_Pool(thread_pool_t *tpt)
{
    long num = 0;
    thread_rwlock_rdlock(&tpt->task_rwlock);
    num = Thread_Get_Task_Number(&tpt->queue);
    (void) thread_rwlock_unlock(&tpt->task_rwlock);
    return num;
}

static void interrupt_call(void *args, thread_task_args_t *args_t) {
    args_t->ttt->handler(args, args_t);
}

int Thread_Pool_Commit(thread_pool_t *tpt)
{
    int err;
    noder node;
    pthread_t tid;
    pthread_attr_t attr;
    thread_container_t *tct;
    if ((err = pthread_attr_init(&attr) != 0))
    {
        fprintf(stderr, "pthread_attr_init failed.\n");
    }
    if ((err = pthread_attr_setschedpolicy(&attr, SCHED_RR) != 0))
    {
        if(err == ENOTSUP)
        {
            fprintf(stdout, "pthread_attr_setschedpolicy SCHED_RR no suppert.\n");
        }else{
            fprintf(stderr, "pthread_attr_setschedpolicy SCHED_RR failed.\n");
        }
    }

    for (int i = 0; i < (tpt->initthreads > tpt->maxthreads?tpt->maxthreads:tpt->initthreads); ++i) {
        thread_mutex_lock(&tpt->tlock);
        err = pthread_create(&tid, &attr, thread_cycle, tpt);
        if (err)
            return thread_err;
    }
    pthread_attr_destroy(&attr);
}
static void cleanup_handler(thread_task_args_t *args) {        // 线程终止后的处理函数防止pthread_cond_wait占用锁导致程序死锁
    thread_pool_t *tpt = args->tpt;
    if(ISNOTNULL(args->tct)) {
        free(args->tct);
        args->tct = NULL;
    }
    if(ISNOTNULL(args)){
        free(args);
        args = NULL;
    }
    thread_mutex_unlock(&tpt->mtx);
}

static inline void *thread_cycle(void *data){
    int err;
    thread_task_args_t *args_t;
    thread_pool_t *tpt = (thread_pool_t*)data;
    pthread_t tid;
    noder node;
    thread_container_t *tct;
    pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
    pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
    tid = pthread_self();
    char ttid[sizeof(char) * tp_numlen(tid)+1];
    memset(ttid, 0, sizeof(ttid));
    tp_sprintf(ttid, "%l", tid);
    thread_rwlock_wrlock(&tpt->thread_rwlock);
    thread_mutex_unlock(&tpt->tlock);
    node = thread_rb_sel_node(&tpt->leader_thread, ttid);
    if(ISNULL(node)){
        args_t = (thread_task_args_t *) tp_calloc(sizeof(thread_task_args_t));
        tct = (thread_container_t *) tp_calloc(sizeof(thread_container_t));
        tct->tid = tid;
        tct->runstate = TP_THREAD_STATE_RUNNABLE;
        tct->freeresiduetimeout = tp_times();              // 这里其实应该是在运行状态 但是为了满足新增线程都是空闲这一策略 依然赋予空闲状态， 很快该状态被修改为运行态
        thread_rb_add_node(tpt, &tpt->leader_thread, ttid, tct);
    } else {
        tct = (thread_container_t *)node->spot->value;
        if(tct->runstate != TP_THREAD_STATE_RUNNABLE){
            tct->runstate = TP_THREAD_STATE_RUNNABLE;
            thread_rb_mod_node(&tpt->leader_thread, ttid, tct);
        }
    }
    args_t->tpt = tpt;
    args_t->tct = tct;
    thread_rwlock_unlock(&tpt->thread_rwlock);
    for (;;) {
        while (!hasTask(tpt)) {                  // 防止线程被意外唤醒
            thread_rwlock_wrlock(&tpt->thread_rwlock);
            tct->freeresiduetimeout = tp_times();          // 由于没有找到任务立刻将当前线程设为空闲状态并赋予时间戳
            node = thread_rb_sel_node( &tpt->leader_thread, ttid);
            if(ISNULL(node))
            {
                (void) thread_rwlock_unlock(&tpt->thread_rwlock);
                return NULL;
            }
            tct = (thread_container_t *)node->spot->value;
            if(tct->runstate != TP_THREAD_STATE_BLOCKED){
                tct->runstate = TP_THREAD_STATE_BLOCKED;
                thread_rb_mod_node(&tpt->leader_thread, ttid, tct);
            }
            (void) thread_rwlock_unlock(&tpt->thread_rwlock);
            pthread_cleanup_push(cleanup_handler, args_t);
            thread_mutex_lock(&tpt->mtx);
wait:       if (pthread_cond_wait(&tpt->cond, &tpt->mtx)) //如果在调用线程尚未锁定互斥锁的情况下调用pthread_cond_wait()，则程序具有未定义的行为。 任何事情都可能发生，包括崩溃。
            {
                (void) thread_mutex_unlock(&tpt->mtx);
                return NULL;
            }
            (void) thread_mutex_unlock(&tpt->mtx);
            pthread_cleanup_pop(0);
            thread_rwlock_wrlock(&tpt->thread_rwlock);
            if(tct->runstate == TP_THREAD_STATE_TERMINATED)
            {
                printf("thread is dead!");
                pthread_cond_signal(&tpt->cond);
                goto wait;
            }
            if(tct->runstate != TP_THREAD_STATE_RUNNABLE){
                tct->runstate = TP_THREAD_STATE_RUNNABLE;
                thread_rb_mod_node(&tpt->leader_thread, ttid, tct);
            }
            tct->freeresiduetimeout = -1;               // 空闲时间戳为-1表示当前线程正在使用中
            (void) thread_rwlock_unlock(&tpt->thread_rwlock);
        }
        thread_task_t *task = Thread_Task_Pop(tpt); // 从队列中取出一个任务
        if(ISNULL(task)) continue;
        args_t->ttt = task;
        args_t->args = task->ctx;
        interrupt_call(task->ctx, args_t);
        Thread_Task_Destry(tpt, task);
    }
}

static inline table_t * select_one_thread_outtime(noder desn, thread_pool_t *tpt){
    thread_rwlock_wrlock(&tpt->thread_rwlock);
    linklisteachinit(&tpt->list_thread);
    while(hasnext(&tpt->list_thread))
    {
        table_t *t = getTb(&tpt->list_thread);
        noder node = RBtreenodesel(&tpt->leader_thread, t->value);
        long time = tp_times();
        if((((thread_container_t *)node->spot->value)->freeresiduetimeout != -1)
           &&(time - ((thread_container_t *)node->spot->value)->freeresiduetimeout) > tpt->threadresiduetimes
           && ((thread_container_t *)node->spot->value)->runstate == TP_THREAD_STATE_BLOCKED)
        {
            ((thread_container_t *)node->spot->value)->runstate = TP_THREAD_STATE_TERMINATED;
            treenodecopy(desn, node);
            (void) thread_rwlock_unlock(&tpt->thread_rwlock);
            return t;
        }
    }
    (void) thread_rwlock_unlock(&tpt->thread_rwlock);
}
void *Thread_Pool_Monitor_Daemon(thread_pool_t *tpt)
{
    int err;
    noder node;
    int res_kill;
    pthread_t tid;
    pthread_attr_t attr;
    thread_container_t *tct;
    noder dnode = (noder)malloc(sizeof(nod));

    if ((err = pthread_attr_init(&attr) != 0))
    {
        fprintf(stderr, "pthread_attr_init failed.\n");
    }
    if ((err = pthread_attr_setschedpolicy(&attr, SCHED_RR) != 0))
    {
        if(err == ENOTSUP)
        {
            fprintf(stdout, "pthread_attr_setschedpolicy SCHED_RR no suppert.\n");
        }else{
            fprintf(stderr, "pthread_attr_setschedpolicy SCHED_RR failed.\n");
        }
    }
    for(;;){
        if(Thread_Get_Task_Number_From_Pool(tpt) * THREAD_FACTOR > thread_get_threads_num(tpt))       // 当线程数量不够时
        {
            thread_mutex_lock(&tpt->tlock);
            if(tpt->maxthreads > thread_get_threads_num(tpt)) {
                err = pthread_create(&tid, &attr, thread_cycle, tpt);
                if (err){
                    thread_mutex_unlock(&tpt->tlock);
                    return NULL;
                }
            }else thread_mutex_unlock(&tpt->tlock);
        }
next:   if(Thread_Get_Task_Number_From_Pool(tpt) == 0 && tpt->minthreads < tpt->threads)  // 当没有任务并且允许的最小线程数小于当前线程数时
        {
            treenodenull(dnode);
            table_t *t = select_one_thread_outtime(dnode, tpt);
            if (ISNULL(dnode)|| ISNULL(dnode->key) || ISNULL(dnode->spot)) continue;
            thread_container_t *ct = (thread_container_t *) dnode->spot->value;
            if (ISNULL(ct)) continue;
            res_kill = pthread_kill(ct->tid, 0);
            if (res_kill == ESRCH){
                fprintf(stderr, "the specified thread did not exists or already quit\n");
            }
            else if (res_kill == EINVAL)
                fprintf(stderr, "signal is invalid\n");
            else {
                int res = pthread_cancel(ct->tid);
                if (res != 0) {
                    fprintf(stderr, "stop thread-%d fail!\n", ct->tid);
                } else {
                    thread_rb_del_node_safe(tpt, &tpt->leader_thread, dnode->key, t);
                    fprintf(stdout, "stop thread-%d success!\n", ct->tid);
                }
            }
        }
    }
}

pthread_t Thread_Pool_Monitor(thread_pool_t *tpt){
    pthread_t tid;
    pthread_create(&tid,NULL,(void *(*)(void *))Thread_Pool_Monitor_Daemon,(void *)tpt);
    return tid;
}

void Thread_Pool_Init(thread_pool_t* tpt)
{
    initLeader(&tpt->leader_thread);
    initLinkList(&tpt->list_thread);
    tpt->name = "ta&to";
    tpt->initthreads = THREAD_INIT_NUMBER;
    tpt->maxthreads = THREAD_MAX_NUMBER;
    tpt->minthreads = THREAD_MIN_NUMBER;
    tpt->threads = 0;
    tpt->waiting = 0;
    tpt->max_queue = THREAD_TASK_MAX_QUEUE;
    tpt->threadresiduetimes = THREAD_RESIDUE_TIMES;
    //tpt->mtx = PTHREAD_MUTEX_INITIALIZER;
    pthread_mutex_init(&tpt->mtx, NULL);
    pthread_mutex_init(&tpt->tlock, NULL);
    //tpt->thread_rwlock = PTHREAD_RWLOCK_INITIALIZER;
    pthread_rwlock_init(&tpt->thread_rwlock, NULL);
    //tpt->task_rwlock = PTHREAD_RWLOCK_INITIALIZER;
    pthread_rwlock_init(&tpt->task_rwlock, NULL);
    //tpt->cond = PTHREAD_COND_INITIALIZER;
    pthread_cond_init(&tpt->cond, NULL);
    Thread_Create_task_queue(&tpt->queue);
}